package com.sunzm.flink.callrecordconnect

import com.alibaba.fastjson.JSON
import org.apache.commons.lang3.StringUtils
import org.apache.flink.api.common.functions.FlatMapFunction
import org.apache.flink.api.common.state.{ValueState, ValueStateDescriptor}
import org.apache.flink.api.common.time.Time
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.co.{KeyedCoProcessFunction, RichCoFlatMapFunction}
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector
import org.slf4j.{Logger, LoggerFactory}

/**
 * /**
 * * 双流连接（connect），
 * * 通话记录和满意度消息拼接需求
 * * 使用更加底层的Process Function来实现
 * *
 * */
 */
object CallRecordConnectJobV3 {
  private val logger: Logger = LoggerFactory.getLogger(this.getClass.getName.stripSuffix("$"))
  private val isLocal = true

  def main(args: Array[String]): Unit = {
    //1.创建执行的环境
    val env: StreamExecutionEnvironment = if (isLocal) {
      StreamExecutionEnvironment.createLocalEnvironmentWithWebUI()
    } else {
      StreamExecutionEnvironment.getExecutionEnvironment
    }

    //获取通话记录消息
    val callLogDS: DataStream[String] = env.socketTextStream("82.156.210.70", 9999)

    //解析名keyBy
    val keyedCallLogDS: KeyedStream[String, String] = callLogDS.flatMap(new FlatMapFunction[String, String] {
      override def flatMap(value: String, out: Collector[String]): Unit = {

        if (StringUtils.isNotBlank(value)) {

          try {
            //解析JSON
            val jSONObject = JSON.parseObject(value)

            val callRecordId = jSONObject.getString("callRecordId")
            if (StringUtils.isNotBlank(callRecordId)) {

              out.collect(value)

            } else {
              //打印告警信息
              logger.warn(s"缺少 callRecordId 字段: ${value}")
            }
          } catch {
            case e: Throwable => {
              logger.error(s"JSON解析异常: ${value}", e)
            }
          }

        }

      }
    }).keyBy(line => {
      val jSONObject = JSON.parseObject(line)

      jSONObject.getString("callRecordId")
    })

    //获取满意度消息
    val satisfactionDS: DataStream[String] = env.socketTextStream("82.156.210.70", 8888)

    val keyedSatisfactionDS: KeyedStream[String, String] = satisfactionDS.flatMap(new FlatMapFunction[String, String] {
      override def flatMap(value: String, out: Collector[String]): Unit = {

        if (StringUtils.isNotBlank(value)) {

          try {
            //解析JSON
            val jSONObject = JSON.parseObject(value)

            val callId = jSONObject.getString("callId")
            if (StringUtils.isNotBlank(callId)) {

              out.collect(value)

            } else {
              //打印告警信息
              logger.warn(s"缺少 callId 字段: ${value}")
            }
          } catch {
            case e: Throwable => {
              logger.error(s"JSON解析异常: ${value}", e)
            }
          }

        }

      }
    }).keyBy(line => {
      val jSONObject = JSON.parseObject(line)

      jSONObject.getString("callId")
    })


    //将2个流进行连接
    val connecedDS: ConnectedStreams[String, String] = keyedCallLogDS.connect(keyedSatisfactionDS)

    /**
     * KeyedCoProcessFunction的泛型说明
     * 第一个是key的类型，这里的key就是通话ID，是String
     * IN1： 第二个参数是第一个流中数据的类型
     * IN2： 第三个参数是第二个流中的数据类型
     * 最后一个参数是返回值类型
     */
    val resultDS: DataStream[String] = connecedDS.process(new KeyedCoProcessFunction[String, String, String, String] {

      /**
       * 记录通话记录的状态
       */
      private var callLogState: ValueState[String] = _

      /**
       * 记录满意度评价的状态
       */
      private var satisfactionState: ValueState[String] = _

      /**
       * 用来保存定时器时间的状态
       */
      private var timerState: ValueState[Long] = _

      /**
       * 这个算子（KeyedCoProcessFunction）被创建的时候调用一次
       *
       * @param parameters
       */
      override def open(parameters: Configuration): Unit = {

        callLogState = getRuntimeContext.getState(
          new ValueStateDescriptor[String]("callLogState", classOf[String])
        )

        satisfactionState = getRuntimeContext.getState(
          new ValueStateDescriptor[String]("satisfactionState", classOf[String])
        )

        timerState = getRuntimeContext.getState(
          new ValueStateDescriptor[Long]("timerState", classOf[Long])
        )
      }

      /**
       * 处理第一个流的数据的方法（每来一条数据都会调用一次）
       *
       * @param value
       * @param ctx
       * @param out
       */
      override def processElement1(value: String,
                                   ctx: KeyedCoProcessFunction[String, String, String, String]#Context,
                                   out: Collector[String]): Unit = {
        //收到通话记录数据
        //先判断是否已经收到满意度评价消息
        val satisfactionJSONStr = satisfactionState.value()
        if (StringUtils.isNotEmpty(satisfactionJSONStr)) {
          //说明已经收到满意度评价了
          //开始拼接消息
          val callLogJSON = JSON.parseObject(value)
          val satisfactionJSON = JSON.parseObject(satisfactionJSONStr)

          //获取满意度评分
          val score = satisfactionJSON.getIntValue("score")
          // 清空满意度JSON对象
          satisfactionJSON.clear()

          //添加到通话记录中
          callLogJSON.put("score", score)

          //输出结果
          out.collect(callLogJSON.toJSONString)

          //通话记录JSON也清空一下
          callLogJSON.clear()

          //清空满意度的状态数据
          satisfactionState.clear()

          //获取定时器的时间戳
          val timerTimeMillis = timerState.value()

          //再删除定时器
          ctx.timerService().deleteProcessingTimeTimer(timerTimeMillis)
        } else {
          //没有收到满意度消息
          //把收到的通话记录消息保存到状态中
          callLogState.update(value)

          //注册一个一分钟的定时器
          //获取当前时间的毫秒数时间戳
          val currentTimeMillis = System.currentTimeMillis()
          //注册一个基于处理时间的1分钟定时器
          val timerTimeMillis = currentTimeMillis + Time.minutes(1).toMilliseconds
          ctx.timerService().registerProcessingTimeTimer(timerTimeMillis)
          //把触发定时器执行的时间戳保存到状态中
          timerState.update(timerTimeMillis)
        }

      }

      /**
       * 处理第二个流的数据的方法（每来一条数据都会调用一次）
       *
       * @param value
       * @param ctx
       * @param out
       */
      override def processElement2(value: String,
                                   ctx: KeyedCoProcessFunction[String, String, String, String]#Context,
                                   out: Collector[String]): Unit = {
        //收到满意度评价数据
        //先判断是否已经收到通话记录消息
        val callLogJSONStr = callLogState.value()
        if (StringUtils.isNotEmpty(callLogJSONStr)) {
          //说明已经收到通话记录了
          //开始拼接消息
          val callLogJSON = JSON.parseObject(callLogJSONStr)
          val satisfactionJSON = JSON.parseObject(value)

          //获取满意度评分
          val score = satisfactionJSON.getIntValue("score")
          // 清空满意度JSON对象
          satisfactionJSON.clear()

          //添加到通话记录中
          callLogJSON.put("score", score)

          //输出结果
          out.collect(callLogJSON.toJSONString)

          //通话记录JSON也清空一下
          callLogJSON.clear()

          //把通话记录的状态清空
          callLogState.clear()

          //获取定时器的时间戳
          val timerTimeMillis = timerState.value()

          //再删除定时器
          ctx.timerService().deleteProcessingTimeTimer(timerTimeMillis)
        } else {
          //没有收到通话记录消息
          //把收到的满意度消息保存到状态中
          satisfactionState.update(value)

          //注册一个一分钟的定时器
          //获取当前时间的毫秒数时间戳
          val currentTimeMillis = System.currentTimeMillis()
          //注册一个基于处理时间的1分钟定时器
          val timerTimeMillis = currentTimeMillis + Time.minutes(1).toMilliseconds
          ctx.timerService().registerProcessingTimeTimer(timerTimeMillis)
          //把触发定时器执行的时间戳保存到状态中
          timerState.update(timerTimeMillis)
        }

      }


      /**
       * 定时器被触发的时候执行
       *
       * @param timestamp
       * @param ctx
       * @param out
       */
      override def onTimer(timestamp: Long,
                           ctx: KeyedCoProcessFunction[String, String, String, String]#OnTimerContext,
                           out: Collector[String]): Unit = {

        //获取状态数据
        val callLogJSONStr = callLogState.value()
        val satisfactionJSONStr = satisfactionState.value()
        //先把收到的状态数据发出

        //通话记录
        if (StringUtils.isNotBlank(callLogJSONStr)) {
          logger.warn(s"通话记录数据超时: ${callLogJSONStr}")
          //如果通话记录不是空，就把通话记录发出
          out.collect(callLogJSONStr)
          //清空状态
          callLogState.clear()
        }

        //满意度
        if (StringUtils.isNotBlank(satisfactionJSONStr)) {
          logger.warn(s"满意度数据超时: ${satisfactionJSONStr}")
          //如果满意度不是空，就把满意度发出
          out.collect(satisfactionJSONStr)
          //清空状态
          satisfactionState.clear()
        }

        //把记录定时器时间的状态清空
        timerState.clear()
      }

      /**
       * 程序退出的时候，这个方法会被调用一次
       */
      override def close(): Unit = {

        //程序退出不能清空状态，不然重启后状态数据就无法恢复了

      }
    })

    resultDS.print()

    env.execute(this.getClass.getSimpleName.stripSuffix("$"))
  }
}
